# -*- coding: utf-8 -*-
from datetime import timedelta
from time import time
from utils.operators.rich_sql_sensor import RichSqlSensor
from jms_dm_duty.dm.dm_whole_decide_duty_base_dt import jms_dm__dm_whole_decide_duty_base_dt

label = f"dm_whole_decide_duty_base_dt_{int(time())}"
timeout = timedelta(minutes=60).seconds
doris_jms_dm__dm_whole_decide_duty_base_dt = RichSqlSensor(
    task_id='doris_jms_dm__dm_whole_decide_duty_base_dt',
    pool='broker_load_pool',
    email=['suning@jtexpress.com','yl_bigdata@yl-scm.com'],
    conn_id='doris_uat',
    pre_sql=f"""
    TRUNCATE TABLE jms_dm.dm_whole_decide_duty_base_dt PARTITION (
         p{{{{ execution_date | date_add(-5) | cst_ds_nodash }}}} 
        ,p{{{{ execution_date | date_add(-4) | cst_ds_nodash }}}}
        ,p{{{{ execution_date | date_add(-3) | cst_ds_nodash }}}}
        ,p{{{{ execution_date | date_add(-2) | cst_ds_nodash }}}}
        ,p{{{{ execution_date | date_add(-1) | cst_ds_nodash }}}}
        ,p{{{{ execution_date | date_add(+0) | cst_ds_nodash }}}}
        ,p{{{{ execution_date | date_add(+1) | cst_ds_nodash }}}}
        ,p{{{{ execution_date | date_add(+2) | cst_ds_nodash }}}}
        ,p{{{{ execution_date | date_add(+3) | cst_ds_nodash }}}}
        ,p{{{{ execution_date | date_add(+4) | cst_ds_nodash }}}}
        ,p{{{{ execution_date | date_add(+5) | cst_ds_nodash }}}}
    );
    LOAD LABEL jms_dm.{label} (
        DATA INFILE(
            "hdfs://{{{{var.value.hadoop_namespace_uat}}}}/dw/hive/jms_tmp.db/external/dm_whole_decide_duty_base_dt/dt={{{{ execution_date | date_add(-5) | cst_ds }}}}/*",
            "hdfs://{{{{var.value.hadoop_namespace_uat}}}}/dw/hive/jms_tmp.db/external/dm_whole_decide_duty_base_dt/dt={{{{ execution_date | date_add(-4) | cst_ds }}}}/*",
            "hdfs://{{{{var.value.hadoop_namespace_uat}}}}/dw/hive/jms_tmp.db/external/dm_whole_decide_duty_base_dt/dt={{{{ execution_date | date_add(-3) | cst_ds }}}}/*",
            "hdfs://{{{{var.value.hadoop_namespace_uat}}}}/dw/hive/jms_tmp.db/external/dm_whole_decide_duty_base_dt/dt={{{{ execution_date | date_add(-2) | cst_ds }}}}/*",
            "hdfs://{{{{var.value.hadoop_namespace_uat}}}}/dw/hive/jms_tmp.db/external/dm_whole_decide_duty_base_dt/dt={{{{ execution_date | date_add(-1) | cst_ds }}}}/*",
            "hdfs://{{{{var.value.hadoop_namespace_uat}}}}/dw/hive/jms_tmp.db/external/dm_whole_decide_duty_base_dt/dt={{{{ execution_date | date_add(+0) | cst_ds }}}}/*",
            "hdfs://{{{{var.value.hadoop_namespace_uat}}}}/dw/hive/jms_tmp.db/external/dm_whole_decide_duty_base_dt/dt={{{{ execution_date | date_add(+1) | cst_ds }}}}/*",
            "hdfs://{{{{var.value.hadoop_namespace_uat}}}}/dw/hive/jms_tmp.db/external/dm_whole_decide_duty_base_dt/dt={{{{ execution_date | date_add(+2) | cst_ds }}}}/*",
            "hdfs://{{{{var.value.hadoop_namespace_uat}}}}/dw/hive/jms_tmp.db/external/dm_whole_decide_duty_base_dt/dt={{{{ execution_date | date_add(+3) | cst_ds }}}}/*",
            "hdfs://{{{{var.value.hadoop_namespace_uat}}}}/dw/hive/jms_tmp.db/external/dm_whole_decide_duty_base_dt/dt={{{{ execution_date | date_add(+4) | cst_ds }}}}/*",
            "hdfs://{{{{var.value.hadoop_namespace_uat}}}}/dw/hive/jms_tmp.db/external/dm_whole_decide_duty_base_dt/dt={{{{ execution_date | date_add(+5) | cst_ds }}}}/*"
        )
        INTO TABLE dm_whole_decide_duty_base_dt
        FORMAT AS 'PARQUET'(
             waybill_no
            ,platform_effect
            ,ziyou_platform_effect
            ,if_platform_standard
            ,if_doudi_platform_standard
            ,cainiao_platform_effect
            ,if_cainiao_platform_pres_standard
            ,order_source_code
            ,order_source_name
            ,duty_type
            ,duty_main_code
            ,duty_main_name
            ,duty_main_type
            ,duty_main_fran_code
            ,duty_main_fran_name
            ,duty_main_agent_code
            ,duty_main_agent_name
            ,duty_provider_code
            ,duty_provider_desc
            ,duty_shipment_no
            ,duty_shipment_name
            ,qxt_dynamic_scan_time
            ,qxt_static_scan_time
            ,qxt_scan_site_actual_arrival_time
            ,qxt_scan_site_plan_arrival_time
            ,ziyou_duty_type
            ,ziyou_duty_main_code
            ,ziyou_duty_main_name
            ,ziyou_duty_main_type
            ,ziyou_duty_main_fran_code
            ,ziyou_duty_main_fran_name
            ,ziyou_duty_main_agent_code
            ,ziyou_duty_main_agent_name
            ,ziyou_duty_provider_code
            ,ziyou_duty_provider_desc
            ,ziyou_duty_shipment_no
            ,ziyou_duty_shipment_name
            ,zy_dynamic_scan_time
            ,zy_static_scan_time
            ,zy_scan_site_actual_arrival_time
            ,zy_scan_site_plan_arrival_time
            ,start_network_code
            ,start_network_name
            ,start_city_code
            ,start_city_name
            ,start_provider_code
            ,start_provider_name
            ,start_fran_code
            ,start_fran_name
            ,start_agent_code
            ,start_agent_name
            ,start_center_code
            ,start_center_name
            ,end_network_code
            ,end_network_name
            ,end_city_code
            ,end_city_name
            ,end_provider_code
            ,end_provider_name
            ,end_fran_code
            ,end_fran_name
            ,end_agent_code
            ,end_agent_name
            ,end_center_code
            ,end_center_name
            ,cainiao_plan_sign_time
            ,ziyou_plan_sign_time
            ,aging_sign_time
            ,actual_sign_time
            ,customer_code
            ,customer_name
            ,duty_main_bagg_user_code
            ,duty_main_bagg_user_name
            ,duty_main_pre_site_code
            ,duty_main_pre_site_name
            ,duty_main_pre_site_type
            ,duty_main_next_site_code
            ,duty_main_next_site_name
            ,duty_main_next_site_type
            ,ziyou_duty_main_bagg_user_code
            ,ziyou_duty_main_bagg_user_name
            ,ziyou_duty_main_pre_site_code
            ,ziyou_duty_main_pre_site_name
            ,ziyou_duty_main_pre_site_type
            ,ziyou_duty_main_next_site_code
            ,ziyou_duty_main_next_site_name
            ,ziyou_duty_main_next_site_type
            ,abnormal_type_code    
            ,abnormal_type_name    
            ,duty_city_code           
            ,duty_city_name           
            ,ziyou_duty_city_code     
            ,ziyou_duty_city_name     
        )
        columns from path as (dt)
        set(
             waybill_no                              =  waybill_no
            ,platform_effect                         =  platform_effect
            ,ziyou_platform_effect                   =  ziyou_platform_effect
            ,if_platform_standard                    =  if_platform_standard
            ,if_doudi_platform_standard              =  if_doudi_platform_standard
            ,cainiao_platform_effect                 =  cainiao_platform_effect
            ,if_cainiao_platform_pres_standard       =  if_cainiao_platform_pres_standard
            ,order_source_code                       =  order_source_code
            ,order_source_name                       =  order_source_name
            ,duty_type                               =  duty_type
            ,duty_main_code                          =  duty_main_code
            ,duty_main_name                          =  duty_main_name
            ,duty_main_type                          =  duty_main_type
            ,duty_main_fran_code                     =  duty_main_fran_code
            ,duty_main_fran_name                     =  duty_main_fran_name
            ,duty_main_agent_code                    =  duty_main_agent_code
            ,duty_main_agent_name                    =  duty_main_agent_name
            ,duty_provider_code                      =  duty_provider_code
            ,duty_provider_desc                      =  duty_provider_desc
            ,duty_shipment_no                        =  duty_shipment_no
            ,duty_shipment_name                      =  duty_shipment_name
            ,qxt_dynamic_scan_time                   =  qxt_dynamic_scan_time
            ,qxt_static_scan_time                    =  qxt_static_scan_time
            ,qxt_scan_site_actual_arrival_time       =  qxt_scan_site_actual_arrival_time
            ,qxt_scan_site_plan_arrival_time         =  qxt_scan_site_plan_arrival_time
            ,ziyou_duty_type                         =  ziyou_duty_type
            ,ziyou_duty_main_code                    =  ziyou_duty_main_code
            ,ziyou_duty_main_name                    =  ziyou_duty_main_name
            ,ziyou_duty_main_type                    =  ziyou_duty_main_type
            ,ziyou_duty_main_fran_code               =  ziyou_duty_main_fran_code
            ,ziyou_duty_main_fran_name               =  ziyou_duty_main_fran_name
            ,ziyou_duty_main_agent_code              =  ziyou_duty_main_agent_code
            ,ziyou_duty_main_agent_name              =  ziyou_duty_main_agent_name
            ,ziyou_duty_provider_code                =  ziyou_duty_provider_code
            ,ziyou_duty_provider_desc                =  ziyou_duty_provider_desc
            ,ziyou_duty_shipment_no                  =  ziyou_duty_shipment_no
            ,ziyou_duty_shipment_name                =  ziyou_duty_shipment_name
            ,zy_dynamic_scan_time                    =  zy_dynamic_scan_time
            ,zy_static_scan_time                     =  zy_static_scan_time
            ,zy_scan_site_actual_arrival_time        =  zy_scan_site_actual_arrival_time
            ,zy_scan_site_plan_arrival_time          =  zy_scan_site_plan_arrival_time
            ,start_network_code                      =  start_network_code
            ,start_network_name                      =  start_network_name
            ,start_city_code                         =  start_city_code
            ,start_city_name                         =  start_city_name
            ,start_provider_code                     =  start_provider_code
            ,start_provider_name                     =  start_provider_name
            ,start_fran_code                         =  start_fran_code
            ,start_fran_name                         =  start_fran_name
            ,start_agent_code                        =  start_agent_code
            ,start_agent_name                        =  start_agent_name
            ,start_center_code                       =  start_center_code
            ,start_center_name                       =  start_center_name
            ,end_network_code                        =  end_network_code
            ,end_network_name                        =  end_network_name
            ,end_city_code                           =  end_city_code
            ,end_city_name                           =  end_city_name
            ,end_provider_code                       =  end_provider_code
            ,end_provider_name                       =  end_provider_name
            ,end_fran_code                           =  end_fran_code
            ,end_fran_name                           =  end_fran_name
            ,end_agent_code                          =  end_agent_code
            ,end_agent_name                          =  end_agent_name
            ,end_center_code                         =  end_center_code
            ,end_center_name                         =  end_center_name
            ,cainiao_plan_sign_time                  =  cainiao_plan_sign_time
            ,ziyou_plan_sign_time                    =  ziyou_plan_sign_time
            ,aging_sign_time                         =  aging_sign_time
            ,actual_sign_time                        =  actual_sign_time
            ,customer_code                           =  customer_code
            ,customer_name                           =  customer_name
            ,duty_main_bagg_user_code                =  duty_main_bagg_user_code
            ,duty_main_bagg_user_name                =  duty_main_bagg_user_name
            ,duty_main_pre_site_code                 =  duty_main_pre_site_code
            ,duty_main_pre_site_name                 =  duty_main_pre_site_name
            ,duty_main_pre_site_type                 =  duty_main_pre_site_type
            ,duty_main_next_site_code                =  duty_main_next_site_code
            ,duty_main_next_site_name                =  duty_main_next_site_name
            ,duty_main_next_site_type                =  duty_main_next_site_type
            ,ziyou_duty_main_bagg_user_code          =  ziyou_duty_main_bagg_user_code
            ,ziyou_duty_main_bagg_user_name          =  ziyou_duty_main_bagg_user_name
            ,ziyou_duty_main_pre_site_code           =  ziyou_duty_main_pre_site_code
            ,ziyou_duty_main_pre_site_name           =  ziyou_duty_main_pre_site_name
            ,ziyou_duty_main_pre_site_type           =  ziyou_duty_main_pre_site_type
            ,ziyou_duty_main_next_site_code          =  ziyou_duty_main_next_site_code
            ,ziyou_duty_main_next_site_name          =  ziyou_duty_main_next_site_name
            ,ziyou_duty_main_next_site_type          =  ziyou_duty_main_next_site_type
            ,abnormal_type_code                      =  abnormal_type_code    
            ,abnormal_type_name                      =  abnormal_type_name    
            ,duty_city_code                          =  duty_city_code           
            ,duty_city_name                          =  duty_city_name           
            ,ziyou_duty_city_code                    =  ziyou_duty_city_code     
            ,ziyou_duty_city_name                    =  ziyou_duty_city_name     
            ,dt                                      =  dt
        )
    )
    WITH BROKER '{{{{ var.json.doris_brokers_uat | random_choice }}}}'
    PROPERTIES ('timeout'='{timeout}', 'max_filter_ratio'='0.0')""",
    poke_sql=f"SHOW LOAD FROM jms_dm WHERE label = '{label}' ORDER BY CreateTime DESC LIMIT 1",
    sql_on_kill=f"CANCEL LOAD FROM jms_dm WHERE LABEL = '{label}'",
    success=lambda r: r[2] == 'FINISHED',
    failure=lambda r: (r[2] is not None and r[2] == 'CANCELLED', str(r[7])),
    poke_interval=60,
    execution_timeout=timedelta(seconds=timeout + 120), )

doris_jms_dm__dm_whole_decide_duty_base_dt << jms_dm__dm_whole_decide_duty_base_dt